package org.egomsl.mw.mapreduce;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.egomsl.mw.HadoopRecord;
import org.egomsl.mw.record.Record;
import org.egomsl.mw.record.RecordScanner;
import org.egomsl.mw.region.RegionReader;

import java.io.IOException;

public class ABaseRecordReader extends RecordReader<BytesWritable, HadoopRecord> {
    private RecordScanner regionRecordScanner;
    private HadoopRecord currentRecord;

    @Override
    public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext)
            throws IOException, InterruptedException {

        currentRecord = null;
        regionRecordScanner = null;

        if(inputSplit instanceof ABaseInputSplit) {
            ABaseInputSplit aBaseInputSplit = (ABaseInputSplit)inputSplit;
            Path regionDir = new Path(aBaseInputSplit.getRegionDir());
            FileSystem fs = regionDir.getFileSystem(taskAttemptContext.getConfiguration());
            regionRecordScanner = new RegionReader(fs, regionDir).getScanner(true);
        }
    }

    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        if(regionRecordScanner != null && regionRecordScanner.hashNext()) {
            currentRecord = new HadoopRecord(regionRecordScanner.next());
            return true;
        }else {
            currentRecord = null;
        }
        return false;
    }

    @Override
    public BytesWritable getCurrentKey() throws IOException, InterruptedException {
        if(currentRecord != null) {
            return currentRecord.getKey();
        }
        return null;
    }

    @Override
    public HadoopRecord getCurrentValue() throws IOException, InterruptedException {
        if(currentRecord != null) {
            return currentRecord;
        }
        return null;
    }

    @Override
    public float getProgress() throws IOException, InterruptedException {
        return 0;
    }

    @Override
    public void close() throws IOException {

    }
}
